package com.leizhi.demo.monitor.achieve.listener;

import com.leizhi.demo.monitor.achieve.PushStreamService;
import com.leizhi.demo.monitor.pojo.FrameData;
import com.leizhi.demo.monitor.pojo.PushStreamProcessInfo;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.handler.timeout.IdleStateEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.OutputStream;
import java.util.List;

import static com.leizhi.demo.monitor.achieve.RtpService.ATTR_KEY_SSRC;
import static com.leizhi.demo.monitor.utils.SipConstantUtil.TCP;
import static com.leizhi.demo.monitor.utils.SipConstantUtil.UDP;
import static io.netty.handler.timeout.IdleState.READER_IDLE;

/**
 * @Description RTP 监听器（处理部分）
 * @Version 1.0.0-SNAPSHOT
 * @Author leizhi
 * @Date 2025/4/27 10:22
 */
public class RtpHandlerListener extends SimpleChannelInboundHandler<FrameData> {

    private static final Logger logger = LoggerFactory.getLogger(RtpHandlerListener.class);

    private final PushStreamService pushStreamService;
    private final String pushRtmpAddress;
    /* 重启次数 */
    private Integer restart = 1;


    public RtpHandlerListener(PushStreamService pushStreamService, String pushRtmpAddress) {
        this.pushStreamService = pushStreamService;
        this.pushRtmpAddress = pushRtmpAddress;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, FrameData frameData) throws Exception {
        PushStreamProcessInfo processInfo = startFfmpegProcess(frameData.getSsrc(), ctx.channel());
        if (processInfo == null) return;
        switch (processInfo.getFlowProtocol()) {
            case UDP:
                udpChannelRead(processInfo, frameData.getComposite());
                break;
            case TCP:
                tcpChannelRead(processInfo, frameData.getNalUnits());
                break;
            default:
                logger.error("RTP 协议类型错误：{}", processInfo.getFlowProtocol());
                break;
        }
    }

    private void udpChannelRead(PushStreamProcessInfo processInfo, ByteBuf composite) throws IOException {
        try {
            composite.readBytes(processInfo.getFfmpegInput(), composite.readableBytes());
            processInfo.setNewPayload(true);
        } catch (Exception e) {
            logger.error("UDP RTP 处理器异常，错误信息：{}", e.getMessage());
        } finally {
            composite.release();
            processInfo.flush();
        }
    }

    private void tcpChannelRead(PushStreamProcessInfo processInfo, List<ByteBuf> nalUnits) throws IOException {
        OutputStream ffmpegInput = processInfo.getFfmpegInput();
        nalUnits.forEach(nalUnit -> {
            try {
                byte[] nalBytes = new byte[nalUnit.readableBytes()];
                nalUnit.readBytes(nalBytes);
                ffmpegInput.write(nalBytes);
            } catch (IOException e) {
                logger.error("RTP TCP 处理器异常，错误信息：{}", e.getMessage());
            }
        });

        ffmpegInput.flush();
        processInfo.setNewPayload(true);
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == READER_IDLE) {
                Long ssrc = ctx.channel().attr(ATTR_KEY_SSRC).get();
                logger.info("SSRC-{}：长时间内未收到数据包", ssrc);

                PushStreamProcessInfo processInfo = pushStreamService.getProcess(ssrc);
                if (processInfo == null) {
                    ctx.close();
                    logger.warn("SSRC-{}：推流实例不存在，该通道已被关闭", ssrc);
                    return;
                }
                if (restart > 3 || !processInfo.isCloud()) {
                    logger.info("SSRC-{}：准备执行关闭推流实例", ssrc);
                    pushStreamService.stop(processInfo.getDeviceId());
                    return;
                }

                try {
                    pushStreamService.sendInvite(processInfo.getDeviceId());
                    logger.info("SSRC-{}：成功发送实时流媒体请求,当前重启次数：{}", ssrc, restart);
                    restart++;
                } catch (Exception e) {
                    logger.error("SSRC-{}：重启推流失败,当前重启次数：{},错误信息：{}", ssrc, restart, e.getMessage());
                }
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        logger.error("Channel 发生异常：", cause);
    }

    /* 获取可用的 FFmpeg 进程信息，没有则创建新的 */
    private PushStreamProcessInfo startFfmpegProcess(Long ssrc, Channel channel) {
        PushStreamProcessInfo pushStreamProcess = pushStreamService.getProcess(ssrc);
        if (pushStreamProcess == null || !pushStreamProcess.isFfmpegInputClosed()) {
            return pushStreamProcess;
        }

        synchronized (RtpHandlerListener.class) {
            if (!pushStreamService.getProcess(ssrc).isFfmpegInputClosed()) {
                logger.info("SSRC-{}：FFmpeg 进程已启动......", ssrc);
                return pushStreamService.getProcess(ssrc);
            }

            try {
                // 启动 FFmpeg 进程
                Process ffmpegProcess = Runtime.getRuntime().exec(getCommand(ssrc));
                OutputStream ffmpegInput = ffmpegProcess.getOutputStream();

                // 判断是 UDP 还是 TCP 连接来选择处理方式
                if (channel instanceof NioDatagramChannel) {
                    pushStreamProcess.setProcessInfo(ffmpegProcess, ffmpegInput);
                } else {
                    // 因为 TCP 必须等设备连接才会生成子通道，所以只能在这里缓存通道信息
                    channel.attr(ATTR_KEY_SSRC).set(ssrc);
                    pushStreamProcess.setProcessInfo(ffmpegProcess, ffmpegInput, channel);
                }

                logger.info("SSRC-{}：开启新的 FFmpeg 进程......", ssrc);
                return pushStreamProcess;
            } catch (IOException e) {
                throw new RuntimeException("启动 FFmpeg 进程失败, " + e.getMessage());
            }
        }
    }

    /* 获取 FFmpeg 推流命令 */
    private String[] getCommand(Long ssrc) {
        return new String[]{
                "ffmpeg",
                "-f", "h264",        // 输入格式为 H.264
                "-i", "-",           // 从标准输入读取数据
                "-c:v", "copy",      // 直接复制视频流，不进行重新编码
                "-f", "flv",         // 输出格式为 FLV
                pushRtmpAddress + ssrc  // RTMP 推流路径
        };
    }

}
